package cn.com.rpc.invoke;

import cn.com.common.*;
import cn.com.common.exception.RpcException;
import cn.com.rpc.NettyClient;
import cn.com.rpc.channel.Channel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 远程rpc调用对象
 *
 * @author jiaming
 */
public class RpcInvoker<T> extends AbstractInvoker<T> {

    protected static final Logger log = LogManager.getLogger();

    private final AtomicInteger index = new AtomicInteger(0);
    private final Channel[] channels;

    @Override
    public Result doInvoke(Invocation invocation, URI uri) {
        Channel currentClient;
        int length = channels.length;
        if (length == 1) {
            currentClient = channels[0];
        } else {
            currentClient = channels[index.incrementAndGet() % length];
        }
        //建立连接
        if (!currentClient.isConnected()) {
            currentClient.connect();
        }

        Integer timeout = Constant.TIMEOUT;
        Request request = new Request();
        Header header = new Header();
        header.setEventType(Constant.ONEWAY);
        header.setStatus(Constant.SUCCESS);
        header.setSerialType(Constant.PROTOBUF_SERIALIZE);
        header.setId(request.newId());
        request.setHeader(header);
        request.setData(invocation);
        Result result;
        //单向发送不需要返回response
        if (uri.getParameter(Constant.DEFAULT_ONEWAY_KEY, Constant.DEFAULT_ONEWAY)) {
            currentClient.send(request, false);
            result = new Result();
        } else {
            Response response = (Response) currentClient.send(request, timeout).get();
            if (Constant.SUCCESS == response.getHeader().getStatus()) {
                //success
                result = (Result) response.getData();
            } else {
                throw new RpcException("invoke exception");
            }
        }
        return result;
    }

    public RpcInvoker(URI uri) {
        super(uri);
        Channel[] channels = getChannels(uri);
        this.channels = channels;
    }

    public RpcInvoker(URI uri, final Channel[] channels) {
        super(uri);
        this.channels = channels;
    }

    private Channel[] getChannels(URI uri) {
        NettyClient nettyClient = new NettyClient(uri);
        Channel[] remoteChannels = new Channel[1];
        remoteChannels[0] = nettyClient;
        return remoteChannels;
    }

    @Override
    public boolean isAvailable() {
        for (Channel client : channels) {
            if (client.isConnected()) {
                return true;
            }
        }
        return false;
    }

    @Override
    public URI getUri() {
        return super.getUri();
    }

    @Override
    public void destory() {
        for (Channel client : channels) {
            client.close();
        }
    }

}